package rdds.spark.examples;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.storage.StorageLevel;

public class LineCount {

	public static void main(String[] args) {

		SparkConf conf = new SparkConf();

		conf.setAppName("WordCounter")//
				.setMaster("local");

		String fileName = "src/main/java/rdds/spark/examples/LineCount.java";

		JavaSparkContext sc = new JavaSparkContext(conf);
		JavaRDD<String> lines = sc.textFile(fileName, 1);

		
		//定义lineLengths作为Map转换的结果 由于惰性，不会立即计算lineLengths
        //第一个参数为传入的内容，第二个参数为函数操作完后返回的结果类型
		JavaRDD<Integer> lineLengths = lines.map(new Function<String, Integer>() {
			private static final long serialVersionUID = 7866892503124559154L;

			public Integer call(String s) {
				System.out.println(s.length()+"\t"+s);
				return s.length();
			}
		});

		//运行reduce  这是一个动作action  这时候，spark才将计算拆分成不同的task，
        //并运行在独立的机器上，每台机器运行他自己的map部分和本地的reducation，并返回结果集给去驱动程序
		int totalLength = lineLengths.reduce(new Function2<Integer, Integer, Integer>() {
			private static final long serialVersionUID = 3497633092573804762L;

			public Integer call(Integer a, Integer b) {
				return a + b;
			}
		});

		System.out.println(totalLength);
		// 为了以后复用 持久化到内存...
		lineLengths.persist(StorageLevel.MEMORY_ONLY());

		sc.close();
	}

}
